package com.xuanwei.schedule.handler.product;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.xuanwei.dao.entity.XwProduct;
import com.xuanwei.schedule.service.ProductService;
import com.xuanwei.xwcommon.consts.MqType;
import com.xuanwei.xwcommon.entity.dto.MessageDto;
import com.xuanwei.xwcommon.enums.MqHandlerType;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.JobHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@JobHandler(value = "RefreshProductJobHandler")
@Component
@Slf4j
public class RefreshProduct extends IJobHandler {

    private AmqpTemplate amqpTemplate;

    private ProductService productService;

    public RefreshProduct(AmqpTemplate amqpTemplate,
                          ProductService productService) {
        this.amqpTemplate = amqpTemplate;
        this.productService = productService;
    }

    private final Integer cpu = Runtime.getRuntime().availableProcessors();

    private final Integer pageSize = 100;

    private ThreadPoolExecutor threadPoolExecutor;

    {
        threadPoolExecutor = new ThreadPoolExecutor(
                cpu,
                2 * cpu,
                0L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(10));
    }

    @Override
    public ReturnT<String> execute(String s) throws Exception {
        log.info("refresh product...");
        for (int i = 0; i < cpu; i++) {
            final int b =i;
            threadPoolExecutor.execute(()->{
                log.info("{}:executor",Thread.currentThread().getName());
                int j = 1;
                while (true) {
                    Page<XwProduct> page = new Page<>(j, pageSize);
                    productService.findPage(page, cpu,b);
                    if (page.getRecords() == null || page.getRecords().size() == 0) {
                        break;
                    }
                    page.getRecords().forEach(item -> {
                        MessageDto<Integer> messageDto = new MessageDto<>();
                        messageDto.setType(MqHandlerType.INSERT);
                        messageDto.setData(item.getId());
                        amqpTemplate.convertAndSend(MqType.PRODUCT, JSON.toJSONString(messageDto));
                    });
                    j++;
                }
            });
        }

        return SUCCESS;
    }

}
